#pragma once
#include "muduo/protoc/codec.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "mq_consumer.hpp"
#include "mq_virtualhost.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_route.hpp"
namespace mq_zxws
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannalRequestPtr = std::shared_ptr<openChannalRequest>;
    using closeChannalRequestPtr = std::shared_ptr<closeChannalRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumerRequestPtr = std::shared_ptr<basicConsumerRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channal{
    public:
        using ptr = std::shared_ptr<Channal>;
        Channal(const std::string& cid, 
                const muduo::net::TcpConnectionPtr& conn, 
                const ProtobufCodecPtr& codec,
                const ConsumerManager::ptr& cmp,
                const VirtualHost::ptr& host,
                const ThreadPool::ptr& pool)
            :_cid(cid)
            ,_conn(conn)
            ,_codec(codec)
            ,_cmp(cmp)
            ,_host(host)
            ,_pool(pool)
        {}
        ~Channal(){
            if(_consumer.get() != nullptr){
                _cmp->remove(_consumer->qname, _consumer->tag);
            }
        }
        // 交换机的新增和删除
        void declareExchange(const declareExchangeRequestPtr& req){
            bool ret = _host->declareExchange(req->exchange_name(), req->type(), req->durable(), req->auto_delete(), req->args());
            return basicResponce(ret, req->rid(), req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr& req){
            _host->deleteExchange(req->exchange_name());
            return basicResponce(true, req->rid(), req->cid());
        }
        // 队列的新增和删除
        void declareQueue(const declareQueueRequestPtr& req){
            bool ret = _host->declareQueue(req->queue_name(), req->durable(), req->exclusive(), req->auto_delete(), req->args());
            if(ret == false) return basicResponce(ret, req->rid(), req->cid());
            _cmp->initComsumerQueue(req->queue_name());// 初始化队列的消费者管理句柄
            return basicResponce(ret, req->rid(), req->cid()); 
        }
        void deleteQueue(const deleteQueueRequestPtr& req){
            _cmp->destroyComsumerQueue(req->queue_name());
            _host->deleteQueue(req->queue_name());
            return basicResponce(true, req->rid(), req->cid());
        }
        // 绑定和解除绑定
        void queueBind(const queueBindRequestPtr& req){
            bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            return basicResponce(ret, req->rid(), req->cid());
        
        }
        void queueUnBind(const queueUnBindRequestPtr& req){
            _host->unBind(req->exchange_name(), req->queue_name());
            return basicResponce(true, req->rid(), req->cid());
        }
        // 消息发布
        void basicPublish(const basicPublishRequestPtr& req){
            // 1、查看交换机是否存在
            Exchange::ptr ep = _host->selectExchange(req->exchange_name());
            if(ep.get() == nullptr){
                DLOG("交换机 %s 不存在\n", req->exchange_name().c_str());
                return basicResponce(false, req->rid(), req->cid());
            }
            // 2、寻找与匹配规则对应的消息队列进行发布
            BasicProperties *properties = nullptr;
            std::string routine_key;
            if(req->has_properties()){
                properties = req->mutable_properties();
                routine_key = properties->routing_key();
            }
            MsgQueueBindingMap mqbp = _host->getExchangeBindings(req->exchange_name());
            for(auto& binding :mqbp){
                // 进行路由匹配
                if(Router::route(ep->type, routine_key, binding.second->_binding_key)){
                    DLOG("routine : %s, binding : %s", routine_key.c_str(), binding.second->_binding_key.c_str());
                    _host->basicPublish(binding.first, properties, req->body());
                    auto task = std::bind(&Channal::publish, this, binding.first);
                    _pool->push(task);
                }
            }
            return basicResponce(true, req->rid(), req->cid());
        }
        // 消息确认
        void basicAck(const basicAckRequestPtr& req){
            _host->basicAck(req->queue_name(), req->message_id());
            return basicResponce(true, req->rid(), req->cid());
        }
        // 订阅消息
        void basicConsumer(const basicConsumerRequestPtr& req){
            // 1、查看是否存在队列
            bool ret = _host->existsQueue(req->queue_name());
            if(ret == false){
                DLOG("订阅消息失败，%s 队列不存在", req->queue_name().c_str());
                return basicResponce(false, req->rid(), req->cid());
            }
            // 2、创建消费者
            auto cb = std::bind(&Channal::callback, this, std::placeholders::_1, 
                        std::placeholders::_2,std::placeholders::_3);
            _cmp->create(req->queue_name(), req->consumer_tag(), req->auto_ack(), cb);
            DLOG("信道 %s 订阅了 %s", req->cid().c_str(), req->queue_name().c_str());
            return basicResponce(true, req->rid(), req->cid());
        }
        // 取消订阅
        void basicCancel(const basicCancelRequestPtr& req){
            _cmp->remove(req->queue_name(), req->consumer_tag());
            return basicResponce(true, req->rid(), req->cid());
        }
    private:
        void callback(const std::string& tag, const BasicProperties* bp, const std::string& body){
            basicConsumerResponce resp;
            resp.set_consumer_tag(tag);
            resp.set_cid(_cid);
            resp.set_body(body);
            if(bp){
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_deliver_mode(bp->deliver_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            DLOG("cid : %s body : %s", _cid.c_str(), body.c_str());
            _codec->send(_conn, resp);
        }
        void basicResponce(bool ok, const std::string& rid, const std::string& cid){
            basicCommonResponce resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
        void publish(const std::string& qname){
            MessagePtr mp = _host->basicConsume(qname);
            if(mp.get() == nullptr){
                DLOG("执行推送消息失败，%s 队列没有消息", qname.c_str());
                return;
            }
            Consumer::ptr cp = _cmp->choose(qname);
            if(cp.get() == nullptr){
                DLOG("执行推送消息失败，%s 队列没有消费者", qname.c_str());
                return;
            }
            DLOG("执行推送消息成功，%s", qname.c_str());
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            if(cp->auto_ack) _host->basicAck(qname, mp->payload().properties().id());
        }
    private:
        std::string _cid;
        Consumer::ptr _consumer; // 用于消费者信道取消订阅时，删除消费信息
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ConsumerManager::ptr _cmp;
        VirtualHost::ptr _host;
        ThreadPool::ptr _pool;
    };

    class ChannalManager{
    public:
        using ptr = std::shared_ptr<ChannalManager>;
        ChannalManager(){}
        bool openChannal(const std::string& cid, 
                const muduo::net::TcpConnectionPtr& conn, 
                const ProtobufCodecPtr& codec,
                const ConsumerManager::ptr& cmp,
                const VirtualHost::ptr& host,
                const ThreadPool::ptr& pool){
            std::unique_lock<std::mutex> lock(_mutex);
            if(_channals.find(cid) != _channals.end()){
                DLOG("打开信道失败，信道 %s 已经存在", cid.c_str());
                return false;
            }
            DLOG("打开信道成功，cid : %s", cid.c_str());
            auto channal = std::make_shared<Channal>(cid, conn, codec, cmp, host, pool);
            _channals.insert(std::make_pair(cid, channal));
            return true;
        }
        void closeChannal(const std::string& id){
            std::unique_lock<std::mutex> lock(_mutex);
            _channals.erase(id);
        }
        Channal::ptr getChannal(const std::string& id){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channals.find(id);
            if(it == _channals.end()){
                DLOG("获取信道失败，信道 %s 不存在", id.c_str());
                return Channal::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channal::ptr> _channals;
    };
}